package loadData

import (
	"fmt"
	"github.com/google/uuid"
	"db2s/global"
	"db2s/outPut"
	"db2s/scheduleTask"
	mq "db2s/topic-mq"
	"reflect"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

// loadOutputMissSegmentQueryListeners loadOutputMissSegmentQuery  监听执行器
func loadOutputMissSegmentQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
		logseq        = message.Properties["logSeq"].(int64)
		dTopic        = loadOutputMissSegmentQuery
		tableSum      int64
		done, done1   = make(chan struct{}, 1), make(chan struct{}, 1)
		event         = "[loadOutputMissSegmentQueryListeners]"
		wg            sync.WaitGroup
		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	plan.Status.SubTaskBeginTime = time.Now().UnixNano()

	if GlobalPConfigs.result.Teletypewriter == "bar" {
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "running")
		WLog.Info(fmt.Sprintf("(%d) %s Add task Bar Run object %v successfully", logseq, event, tableName))
	}

	//处理wheresql
	if v, ok := GlobalPConfigs.schema.WhereSql[fmt.Sprintf("%s%s%s", plan.Object.Schema, global.SchemaTableSplit, plan.Object.Table)]; ok && v != "" {
		tableSum = 1
		plan.whereSqlFilter(&wg, message, v)
	} else {
		tableSum = int64(len(plan.Object.PartitionData.Name.Source))
		plan.missSendMessage(&wg, message)
	}
	wg.Add(3)
	go plan.speedMonitor(done1, &wg)
	//init status
	go mqTableStatusMonitor(done, done1, &wg, plan, logseq)
	go expensiveOperation(&wg, message, done, tableSum, logseq)
	wg.Wait()
	return mq.ConsumeSuccess
}

// loadOutputIsSegmentQueryListeners IsSegmentQuery  监听执行器
func loadOutputIsSegmentQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
		logseq        = message.Properties["logSeq"].(int64)
		dTopic        = loadOutputIsSegmentQuery
		event         = "[loadOutputIsSegmentQueryListeners]"
		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
		wg            sync.WaitGroup
		done          = make(chan struct{}, 1)
		done1         = make(chan struct{}, 1)
		WriteTextI    = outPut.FileOut("text", map[string]any{"fileN": plan.load.WriteFileN, "logSeq": logseq,
			"bufSize": 1024 * 1024 * 4, "rowsS": GlobalPConfigs.rules.Load.RowsExcision, "columnS": GlobalPConfigs.rules.Load.ColumnExcision,
		})
		mqTaskSum int64
	)
	defer deleteTopic(deleteMapInfo)
	pc, _, _, _ := runtime.Caller(1)
	callingFunc := runtime.FuncForPC(pc)
	callFile, callLine := callingFunc.FileLine(pc)
	callFunc := callingFunc.Name()
	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
	plan.Status.SubTaskBeginTime = time.Now().UnixNano()
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "running")
		WLog.Info(fmt.Sprintf("(%d) %v %s Add task Bar Run object %v successfully", logseq, callFuncInfo, event, tableName))
	}
	//处理wheresql
	if v, ok := GlobalPConfigs.schema.WhereSql[fmt.Sprintf("%s%s%s", plan.Object.Schema, global.SchemaTableSplit, plan.Object.Table)]; ok && v != "" {
		mqTaskSum = 1
		plan.whereSqlFilter(&wg, message, v)
	} else {
		mqTaskSum = int64(len(plan.Object.PartitionData.Name.Source))
		plan.IsSegmentSendMessage(&wg, message)
	}
	wg.Add(3)
	go plan.speedMonitor(done1, &wg)
	//init status
	go expensiveOperation(&wg, message, done, mqTaskSum, logseq)
	go mqTableStatusMonitor(done, done1, &wg, plan, logseq)
	go plan.loadOutputWriteData(WriteTextI, logseq)
	// 获取当前计数
	wg.Wait()
	tableMetaWriteFile(plan)
	return mq.ConsumeSuccess
}

// loadOutputIndexConcurrentQueryListeners loadOutputIndexConcurrentQuery 监听执行器
func loadOutputIndexConcurrentQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		logseq        = message.Properties["logSeq"].(int64)
		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
		topic         = loadOutputTableSql
		dTopic        = loadOutputIndexConcurrentQuery
		event         = "[loadOutputIndexConcurrentQueryListeners]"
		fatherSonTags = make(map[string][]string)
		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	segmentSql := message.Properties["segmentSql"].(global.StartPart)
	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 || len(segmentSql.Error) > 0 {
		plan.errorAction(message, logseq, segmentSql.Error)
		return mq.ConsumeSuccess
	}
	//分段查询错误
	//分段查询错误
	taskObject := message.Properties["taskObject"].(fp.TaskDistribution)
	var l = chanSendMsg{
		Object:   "second",
		SendData: taskObject.IsConcurrentValue(segmentSql),
		SendMsgInfo: SendMsgInfo{
			CheckMod:  getCheckMod(),
			Topic:     topic,
			TableName: tableName,
		},
		SendMsg: sendMsg,
		FatherSon: FatherSon{
			FatherTags: map[string]any{"logSeq": logseq,
				"taskObject": message.Properties["taskObject"],
				"sendMsgKey": "sqlwhere", "subTask": plan.subTaskInfo, "label": message.Properties["label"],
				"plan": message.Properties["plan"], "sourcePartitionName": message.Properties["sourcePartitionName"],
				"insertPrefix": message.Properties["insertPrefix"],
				"topic":        string(topic), "outFay": GlobalPConfigs.result.Teletypewriter},
			FatherSonTags: fatherSonTags,
		},
		SendMsgEnd: func() {
		},
	}
	if !chanMonitorSendMsg(&l) {
		return mq.ReconsumeLater
	}
	if atomic.LoadInt64(plan.Status.ErrEnd) > -1 {
		plan.mqProductCustomerMonitor.ProductMessage <- fatherSonTags
		atomic.AddInt64(plan.mqProductCustomerMonitor.MqLevelCustomerReceive, 1)
	}
	return mq.ConsumeSuccess
}

// loadOutputTableSqlListeners QueryTableSql 监听执行器
func loadOutputTableSqlListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
		logseq        = message.Properties["logSeq"].(int64)
		topic         = loadOutputQueryTableData
		dTopic        = loadOutputTableSql
		event         = "[queryTableSqlListeners]"
		tableSql      string
		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	//分段查询错误
	sqlWhere := message.Properties["sqlwhere"].(global.StartPart)
	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 || len(sqlWhere.Error) > 0 {
		plan.errorAction(message, logseq, sqlWhere.Error)
		return mq.ReconsumeLater
	}
	partitionName := message.Properties["sourcePartitionName"].(string)
	if tableSql = plan.tmpSyncTableQuerySql(sqlWhere, partitionName); len(tableSql) == 0 {
		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
		return mq.ConsumeSuccess
	}
	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
		"taskObject": message.Properties["taskObject"], "label": message.Properties["label"],
		"strsql": tableSql, "err": plan.Status.ErrEnd, "sourcePartitionName": message.Properties["sourcePartitionName"],
		"insertPrefix": message.Properties["insertPrefix"],
		"subTask":      plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}

// loadOutputQueryTableDataListeners loadOutputQueryTableData 监听执行器
func loadOutputQueryTableDataListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		stt          any
		logThreadSeq = message.Properties["logSeq"].(int64)
		tableName    = string(message.Body)
		plan         = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
		err          error
		event        = "[loadOutputQueryTableDataListeners]"
		dTopic       = loadOutputQueryTableData
		ssum         int64
		//topic         = loadOutputWriteData
		deleteMapInfo = map[string]any{"logSeq": logThreadSeq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	defer func() {
		atomic.AddInt64(plan.TableSpeedLimit.ReadParallel, -1)
	}()
	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
		return mq.ConsumeSuccess
	}
	querySql := message.Properties["strsql"].(string)
	partitionName := message.Properties["sourcePartitionName"].(string)
	atomic.AddInt64(plan.TableSpeedLimit.ReadParallel, 1)
	if stt, err = plan.syncQueryTableData(querySql, partitionName, logThreadSeq); err != nil {
		plan.errorAction(message, logThreadSeq, err)
		return mq.ConsumeSuccess
	}
	switch fmt.Sprintf("%v", reflect.TypeOf(stt)) {
	case "[]*string":
		if len(stt.([]*string)) == 0 {
			atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
			return mq.ConsumeSuccess
		}
		ssum = int64(len(stt.([]*string)))
	case "[][]*string":
		if len(stt.([][]*string)) == 0 {
			atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
			return mq.ConsumeSuccess
		}
		ssum = int64(len(stt.([][]*string)))
	default:
		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
		return mq.ConsumeSuccess
	}
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		if PlanContext.TaskBarSubsStatusGet(plan.subTaskInfo) != -2 {
			PlanContext.SubTaskBarAccumulate(map[string]any{"subTask": plan.subTaskInfo, "select": ssum, "insert": int64(0)})
		}
	}
	atomic.AddInt64(plan.TableSpeedLimit.ReadSum, ssum)
	atomic.AddInt64(terminalPods[getTableName(plan.Object.Schema, plan.Object.Table)].Load.SelectRows, ssum)
	plan.writeOptimizer.QueueData <- stt
	//if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logThreadSeq, "plan": message.Properties["plan"],
	//	"taskObject": message.Properties["taskObject"], "label": message.Properties["label"],
	//	"insertPrefix": message.Properties["insertPrefix"], "WriteTextI": message.Properties["WriteTextI"],
	//	"chunkData": stt, "err": plan.Status.ErrEnd, "subTask": plan.subTaskInfo, "sourcePartitionName": message.Properties["sourcePartitionName"],
	//	"topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
	//	return mq.ReconsumeLater
	//}
	return mq.ConsumeSuccess
}

// loadOutputWriteDataListeners loadOutputWriteData 监听执行器
func loadOutputWriteDataListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	//var (
	//	tableName     = string(message.Body)
	//	logseq        = message.Properties["logSeq"].(int64)
	//	plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
	//	dTopic        = loadOutputWriteData
	//	event         string
	//	deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	//)
	//defer deleteTopic(deleteMapInfo)
	//defer func() {
	//	atomic.AddInt64(plan.TableSpeedLimit.WriteParallel, -1)
	//}()
	//if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
	//	return mq.ConsumeSuccess
	//}
	//atomic.AddInt64(plan.TableSpeedLimit.WriteParallel, 1)
	//chunkData := message.Properties["chunkData"]
	//WriteTextI := message.Properties["WriteTextI"]
	//insertSum, err := plan.loadOutputWriteData(chunkData, WriteTextI, logseq)
	//if err != nil {
	//	plan.errorAction(message, logseq, err)
	//	return mq.ReconsumeLater
	//}
	//if GlobalPConfigs.result.Teletypewriter == "bar" {
	//	if atomic.LoadInt64(plan.Status.ErrEnd) > -1 {
	//		PlanContext.SubTaskBarAccumulate(map[string]any{"subTask": plan.subTaskInfo, "select": int64(0), "insert": insertSum})
	//	}
	//}
	//atomic.AddInt64(plan.TableSpeedLimit.WriteSum, insertSum)
	//atomic.AddInt64(terminalPods[getTableName(plan.Object.Schema, plan.Object.Table)].Load.InsertRows, insertSum)
	//atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
	return mq.ConsumeSuccess
}

// loadInputFileDataListeners loadInputFileData 监听执行器
func loadInputFileDataListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		logseq        = message.Properties["logSeq"].(int64)
		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
		topic         = loadInputGenerlInsert
		dTopic        = loadInputFileData
		event         = "[loadInputFileDataListeners]"
		fatherSonTags = make(map[string][]string)
		insertPrefix  string
		err           error
		wg            sync.WaitGroup
		done          = make(chan struct{}, 1)
		done1         = make(chan struct{}, 1)
		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	if !tableMetaReadFile(plan) {
		return mq.ReconsumeLater
	}
	defer deleteTopic(deleteMapInfo)
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "running")
		WLog.Info(fmt.Sprintf("(%d) %s Add task Bar Run object %v successfully", logseq, event, tableName))
	}
	if insertPrefix, err = plan.ifTableLobColumn("single"); err != nil {
		plan.errorAction(message, logseq, err)
		return mq.ReconsumeLater
	}
	wg.Add(3)
	//init status
	go expensiveOperation(&wg, message, done, 1, logseq)
	go mqTableStatusMonitor(done, done1, &wg, plan, logseq)
	go plan.speedMonitor(done1, &wg)
	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
		return mq.ReconsumeLater
	}
	readData := plan.loadInputReadData1()
	uid := uuid.New()
	l := chanSendMsg{
		Object:   "first",
		SendData: readData,
		SendMsgInfo: SendMsgInfo{
			CheckMod:  getCheckMod(),
			Topic:     topic,
			TableName: tableName,
		},
		SendMsg:          sendMsg,
		SpeedLimitSwitch: false,
		FatherSon: FatherSon{
			FatherTags: map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
				"sendMsgKey": "rowsData", "label": map[string]any{uid.String(): []string{}},
				"err": plan.Status.ErrEnd, "subTask": plan.subTaskInfo, "topic": string(topic),
				"insertPrefix":        insertPrefix,
				"sourcePartitionName": "single",
				"outFay":              GlobalPConfigs.result.Teletypewriter},
			FatherSonTags: fatherSonTags,
		},
		SendMsgEnd: func() {
		},
	}
	if !chanMonitorSendMsg(&l) {
		return mq.ReconsumeLater
	}
	plan.mqProductCustomerMonitor.ProductMessage <- fatherSonTags
	atomic.SwapInt64(plan.mqProductCustomerMonitor.MqStartTiming, 1)
	atomic.SwapInt64(plan.mqProductCustomerMonitor.MqLevelProductSend, int64(len(fatherSonTags)))
	atomic.SwapInt64(plan.mqProductCustomerMonitor.MqLevelCustomerReceive, int64(len(fatherSonTags)))
	wg.Wait()
	return mq.ConsumeSuccess
}

// loadInputGenerlInsertListeners loadInputGenerlInsert 监听执行器
func loadInputGenerlInsertListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		logseq        = message.Properties["logSeq"].(int64)
		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
		topic         = loadInputWriteInsert
		dTopic        = loadInputGenerlInsert
		event         string
		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	defer func() {
		atomic.AddInt64(plan.TableSpeedLimit.ReadParallel, -1)
	}()
	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
		return mq.ReconsumeLater
	}
	atomic.AddInt64(plan.TableSpeedLimit.ReadParallel, 1)
	rowsData := message.Properties["rowsData"]
	insertPrefix := message.Properties["insertPrefix"].(string)
	insertSql, lobSwitch, err := plan.loadDataInputGeneralInsert1(insertPrefix, rowsData, logseq)
	if err != nil {
		plan.errorAction(message, logseq, err)
		return mq.ConsumeSuccess
	}
	switch fmt.Sprintf("%v", reflect.TypeOf(insertSql)) {
	case "chan interface {}":
		l := chanSendMsg{
			SendData: insertSql,
			SendMsgInfo: SendMsgInfo{
				CheckMod:  getCheckMod(),
				Topic:     topic,
				TableName: tableName,
			},
			SpeedLimitSwitch: false,
			SendMsg:          sendMsg,
			ExecFunc: func() {
				atomic.AddInt64(plan.mqProductCustomerMonitor.MqProductSeq, 1)
			},
			FatherSon: FatherSon{
				FatherTags: map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
					"insertPrefix": message.Properties["insertPrefix"],
					"lobSwitch":    lobSwitch, "sendMsgKey": "insertSql", "err": plan.Status.ErrEnd,
					"subTask":             plan.subTaskInfo,
					"sourcePartitionName": message.Properties["sourcePartitionName"],
					"topic":               string(topic),
					"outFay":              GlobalPConfigs.result.Teletypewriter},
			},
			SendMsgEnd: func() {
				//atomic.AddInt64(plan.mqProductCustomerMonitor.MqProductSeq, -1)
			},
		}
		chanMonitorSendMsg(&l)
	}
	return mq.ConsumeSuccess
}

// loadInputWriteInsertListeners loadInputWriteInsert 监听执行器
func loadInputWriteInsertListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		logseq        = message.Properties["logSeq"].(int64)
		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
		dTopic        = loadInputWriteInsert
		event         string
		insertSum     int64
		insertsql     any
		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	defer func() {
		atomic.AddInt64(plan.TableSpeedLimit.WriteParallel, -1)
		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
	}()
	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
		return mq.ConsumeSuccess
	}
	atomic.AddInt64(plan.TableSpeedLimit.WriteParallel, 1)
	insertSuffix := message.Properties["insertSql"]
	if insertSuffix != nil {
		switch fmt.Sprintf("%v", reflect.TypeOf(insertSuffix)) {
		case "map[int64]*string":
			for k, v := range insertSuffix.(map[int64]*string) {
				insertSum = k
				insertsql = v
			}
		case "map[int64][][]*string":
			for k, v := range insertSuffix.(map[int64][][]*string) {
				insertSum = k
				insertsql = v
			}
		}
	}
	insertPrefix := message.Properties["insertPrefix"].(string)
	lobSwitch := message.Properties["lobSwitch"].(bool)
	sourcePartitionName := message.Properties["sourcePartitionName"].(string)
	if lobSwitch {
		if err := plan.writeInsert(lobSwitch, insertPrefix, insertsql, sourcePartitionName, logseq); err != nil {
			plan.errorAction(message, logseq, err)
			return mq.ConsumeSuccess
		}
	} else {
		if err := plan.writeInsert(lobSwitch, insertPrefix, insertsql, sourcePartitionName, logseq); err != nil {
			plan.errorAction(message, logseq, err)
			return mq.ConsumeSuccess
		}
	}
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		if PlanContext.TaskBarSubsStatusGet(plan.subTaskInfo) != -2 {
			PlanContext.SubTaskBarAccumulate(map[string]any{"subTask": plan.subTaskInfo, "select": int64(0), "insert": insertSum})
		}
	}
	atomic.AddInt64(plan.TableSpeedLimit.WriteSum, insertSum)
	atomic.AddInt64(terminalPods[getTableName(plan.Object.Schema, plan.Object.Table)].Load.InsertRows, insertSum)
	return mq.ConsumeSuccess
}
